package com.hundsun.cprs.yyt.service.match.core;

import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.hundsun.cprs.yyt.service.match.core.command.MatchingCommand;
import com.hundsun.cprs.yyt.service.match.core.command.MatchingCommandQueue;
import com.hundsun.cprs.yyt.service.match.core.queue.PriceQueue;
import com.hundsun.cprs.yyt.service.match.core.queue.TimePriorityCompositeQueue;
import com.hundsun.cprs.yyt.service.match.core.queue.TimePriorityQueue;
import com.hundsun.cprs.yyt.service.match.core.stat.TradesQuotes;

/**
 * 匹配中心
 */
public class Matching {

    private static final Log logger = LogFactory.getLog(Matching.class);

    private MatchingCommandQueue commandQueue = new MatchingCommandQueue();

    private TimePriorityCompositeQueue sellQueue = new TimePriorityCompositeQueue(1,1);//默认新老队列轮询

    private PriceQueue buyQueue = new TimePriorityQueue();
    
    private TradesQuotes quotes ;

    private Thread matchingThread;

    private AtomicInteger matchingThreadSuccessCount = new AtomicInteger(0); // 成功执行过指令数

    private AtomicInteger matchingThreadFailedCount = new AtomicInteger(0); // 指令执行失败数

    private Executor executor;

    private TradeCenter tradeCenter;

    public Matching(TradeCenter tradeCenter, Executor executor) {
        this.tradeCenter = tradeCenter;
        this.executor = executor;
        this.quotes = new TradesQuotes();
    }

    /**
     * 在待撮合队列中增加一个指令
     * 
     * @param command
     */
    public void putCommand(MatchingCommand command) {
        this.commandQueue.addCommand(command);
    }

    /**
     * 启动匹配线程
     */
    public void start() {
        shutdown();
        matchingThread = new Thread(new Runnable() {
            public void run() {
                for (;;) {
                    try {
                        MatchingCommand command = commandQueue.getCommand();
                        if (logger.isDebugEnabled()) {
                            logger.debug("current running command is " + command.getClass().getName());
                        }
                        command.perform(Matching.this, Matching.this.executor);
                        matchingThreadSuccessCount.incrementAndGet();
                    } catch (InterruptedException ie) {
                        if (logger.isDebugEnabled()) {
                            logger.debug("Interrupted,exist matching thread." + "current queue size:"
                                + commandQueue.getQueueSize());
                        }
                        break;
                    } catch (Exception e) {
                        matchingThreadFailedCount.incrementAndGet();
                        logger.error("error then perform command", e);
                    }
                }
            }
        }, "stock[" + tradeCenter.getStockCode() + "]matching");
        matchingThread.start();
    }

    /**
     * 关闭撮合线程,注意,无论待撮合队列中是否有指令,都强行关闭撮合线程
     */
    public void shutdown() {
        if (this.matchingThread != null) {
            this.matchingThread.interrupt();
        }
        matchingThreadSuccessCount.set(0);
        matchingThreadFailedCount.set(0);
    }
    
    /**
     * 指令线程是否已经启动
     * 
     * @return
     */
    public boolean isCommandThreadAlive() {
        if (this.matchingThread == null) {
            return false;
        }
        return this.matchingThread.isAlive();
    }
    
    /**
     * 得到指令线程成功执行完成的指令数
     * 
     * @return
     */
    public int getCommandThreadSuccessCount() {
        return this.matchingThreadSuccessCount.intValue();
    }

    /**
     * 得到指令线程执行失败的指令数
     * 
     * @return
     */
    public int getCommandThreadFailedCount() {
        return this.matchingThreadFailedCount.intValue();
    }

    public MatchingCommandQueue getCommandQueue() {
        return commandQueue;
    }

    public Thread getMatchingThread() {
        return matchingThread;
    }

    public AtomicInteger getMatchingThreadSuccessCount() {
        return matchingThreadSuccessCount;
    }

    public AtomicInteger getMatchingThreadFailedCount() {
        return matchingThreadFailedCount;
    }

    public Executor getExecutor() {
        return executor;
    }

    public TradeCenter getTradeCenter() {
        return tradeCenter;
    }

    public TimePriorityCompositeQueue getSellQueue() {
        return sellQueue;
    }

    public PriceQueue getBuyQueue() {
        return buyQueue;
    }

    public TradesQuotes getQuotes() {
        return quotes;
    }

}
